package com.kenliang.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalUnit;
import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * @Author:LiangTao
 * @Description:
 * @Date: Created in 16:03 2018/12/2.
 */
@Slf4j
public class KafkaTest {



    public static void main(String[] args) {


    }

//    需要保证一个消费者使用一个线程？？怎么保证
    private Map<TopicPartition,OffsetAndMetadata> currentOffsets = new HashMap<>();

    private KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(new Properties());

    public void  testKafka(){
        //生产者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(new Properties());
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>("topic", "key", "value");
        try {
            //同步发送
            kafkaProducer.send(producerRecord).get();
            //异步发送
            kafkaProducer.send(producerRecord, (recordMetadata, exception) -> {
                if (exception != null) {
                    exception.printStackTrace();
                }
            });
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }


        GenericRecord genericRecord = new GenericData.Record(new Schema.Parser().parse(""));
        genericRecord.put("id", 1);


        //消费者

        kafkaConsumer.subscribe(Collections.singletonList("topic"));
        ConsumerRecords<String,String> consumerRecords = kafkaConsumer.poll(Duration.of(100L, ChronoUnit.SECONDS));

        //处理记录
        for(ConsumerRecord<String,String> consumerRecord:consumerRecords){
            currentOffsets.put(new TopicPartition(consumerRecord.topic(),consumerRecord.partition()),
                    new OffsetAndMetadata(consumerRecord.offset()+1,"no metadata"));
            kafkaConsumer.commitAsync(currentOffsets,null);
        }

        //提交偏移量
        kafkaConsumer.commitSync();
        kafkaConsumer.commitAsync(
                (offsets, exception)-> {
                    if(exception!=null){

                    }
                }
        );


        //设置再平衡监听器
        kafkaConsumer.subscribe(Collections.singletonList("topic"),new HandleRebalanceListener());

    }

    /**
     * @Author:LiangTao
     * @Description:再平衡监听器
     * @Date: Created in 16:33 2018/12/8.
     */
    private class HandleRebalanceListener implements ConsumerRebalanceListener {

        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

        }

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            kafkaConsumer.commitSync(currentOffsets);
        }
    }


}
